package org.yuanzheng.window

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, ProcessingTimeSessionWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @since 2021/3/6
 */
object TestSessionWindow {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.streaming.api.scala._

    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.8", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })

    //开窗
    // 以处理时间为衡量标准，如果10秒内没有任何数据输入，就认为会话已经关闭，此时触发统计
    stream.map(log => (log.sid, 1)).keyBy(_._1).window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
    stream.map(log => (log.sid, 1)).keyBy(_._1).window(EventTimeSessionWindows.withGap(Time.seconds(10)))

    streamEnv.execute()

  }

}
